/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.sss.client.DataPutter;
import org.sss.client.SssClient;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.datatype.PackableBytes;

public class PutFile {

  public static void usage() {
    System.out.println("java PutFile [-o gid] filespecs");
    System.out.println("    <filespecs> ::= <filespec>* ;");
    System.out.println("     <filespec> ::= <filename> | -c <count> <filename> ;");
    System.out.println("        <count> ::= [1-9][0-9]* ;");
    System.exit(-1);
  }

  public static byte[] readFile(String pathname) throws IOException {
    File file = new File(pathname);
    FileInputStream fin = new FileInputStream(file);
    BufferedInputStream in = new BufferedInputStream(fin);
    byte[] buffer = new byte[(int)file.length()];

    in.read(buffer);
    in.close();
    fin.close();

    return buffer;
  }

  private static class TargetFile {
    public File file;
    public int count;

    public TargetFile(File file, int count) {
      this.file  = file;
      this.count = count;
    }
  }

  private static class ParsedInfo {
    public List<TargetFile> files = new ArrayList<TargetFile>();
    public GroupID gid = null;
  }

  private static ParsedInfo parseArgs(String[] args) {
    ParsedInfo info = new ParsedInfo();

    for (int i = 0; i < args.length; i++) {
      if (args[i].equals("-o")) {
        if (i+1 >= args.length) {
          System.err.println("runaway argument.");
          return null;
        }
        try {
          if (info.gid == null) {
            info.gid = GroupID.createFromString(args[i + 1]);
          }
          else {
            System.err.println("duplicate options `-o' are specified.");
            return null;
          }
        }
        catch (IllegalArgumentException e) {
          System.err.println("wrong GroupID specified: " + args[i + 1]);
          return null;
        }
        i++;
      }
      else {
        int c = 1;
        if (args[i].equals("-c")) {
          if (i+2 >= args.length) {
            System.err.println("runaway argument.");
            return null;
          }
          try {
            c = Integer.parseInt(args[i + 1]);
            if (c < 1) {
              System.err.println("count should be greater than 0: " + args[i + 1]);
              return null;
            }
          }
          catch (NumberFormatException e) {
            System.err.println("wrong count specified: " + args[i + 1]);
            return null;
          }
          i += 2;
        }
        File file = new File(args[i]);
        if (!file.isFile()) {
          System.err.println("file not found: " + args[i]);
          return null;
        }
        else if (!file.canRead()) {
          System.err.println("file cannot be read: " + args[i]);
          return null;
        }
        info.files.add(new TargetFile(file, c));
      }
    }
    return info;
  }

  /**
   * Usage:
   * java PutFile
   */
  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);

    String[] others = client.getArgs();
    ParsedInfo info = parseArgs(others);
    if (info == null || info.files == null || info.files.isEmpty()) {
      usage();
    }

    Map<String, byte[]> cache = new HashMap<String, byte[]>();

    DataPutter<PackableBytes, PackableBytes> putter =
      DataPutter.create(client, PackableBytes.class, PackableBytes.class, info.gid);
    GroupID table = putter.getOutputGroupID();
    try {
      for (TargetFile file: info.files) {
        String k = file.file.getPath();
        byte[] v = cache.get(k);
        if (v == null) {
          v = readFile(k);
          cache.put(k, v);
        }
        for (int i = 0;i < file.count;++i) {
          putter.append(new PackableBytes(k.getBytes("UTF-8")),
                        new PackableBytes(v));
        }
      }
    } finally {
      putter.close();
    }

    System.out.println("output data deployed - " + table.toString());
  }
}
